﻿/**

 * Copyright (c) 2015-2016, FastDev 刘强 (fastdev@163.com) & Quincy.

 *

 * Licensed under the Apache License, Version 2.0 (the "License");

 * you may not use this file except in compliance with the License.

 * You may obtain a copy of the License at

 *

 *      http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

using ConsoleApplication1.Cluster.Entity;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace OF.Notify.DataHost.Cluster.Disk
{

    public class DiskIdToSmallValuesMapper : IIdToSmallValuesMapper
    {
        internal static int TempId = 0;
        internal const int MaxBufferSize = 1024 * 1024 * 100;
        internal Func<int, string> getIdPathFunc = null;
        internal string basePath = null;
        internal int combineRate = 256;

        internal static int CellLen = 1 + SmallValue16AndFlag.CellLen;
        static DiskIdToSmallValuesMapper()
        {
            CellLen = 1 + SmallValue16AndFlag.CellLen;
        }

        public DiskIdToSmallValuesMapper(string basePath, Func<int, string> getIdPathFunc)
        {
            this.basePath = basePath;
            this.getIdPathFunc = getIdPathFunc;
        }

        internal string GetFilePath(int id)
        {
            return getIdPathFunc(id / combineRate);
        }

        public void AppendValue(int id, SmallValue16AndFlag value)
        {
            string filePath = GetFilePath(id);
            byte idOffset = (byte)(id % combineRate);
            byte[] valueArray = value.GetBytes();
            DiskVisitor.EnsureFileDirectoryExists(filePath);
            ClusterContext.StringLock.Invoke(filePath, () => {
                using (FileStream fs = new FileStream(filePath, FileMode.Append))
                {
                    fs.WriteByte(idOffset);
                    fs.Write(valueArray, 0, valueArray.Length);
                }
            });
        }

        public void AppendValue(int id, List<SmallValue16AndFlag> values)
        {
            string filePath = GetFilePath(id);
            SaveValues(id, filePath, values);
        }

        internal void SaveValues(int id, string filePath, List<SmallValue16AndFlag> values)
        {
            DiskVisitor.EnsureFileDirectoryExists(filePath);
            byte idOffset = (byte)(id % combineRate);
            ClusterContext.StringLock.Invoke(filePath, () => {
                using (FileStream fs = new FileStream(filePath, FileMode.Append))
                {
                    foreach (var value in values)
                    {
                        byte[] btArray = value.GetBytes();
                        fs.WriteByte(idOffset);
                        fs.Write(btArray, 0, btArray.Length);
                    }
                }
            });
        }

        public void ResetValues(int id, List<SmallValue16AndFlag> values)
        {
            string filePath = GetFilePath(id);
            ClusterContext.StringLock.Invoke(filePath, () => {
                if (!File.Exists(filePath))
                {
                    SaveValues(id, filePath, values);
                }
                else
                {
                    byte idOffset = (byte)(id % combineRate);
                    int count = 0;
                    byte[] otherContent = GetFileContent(filePath, (fOffset) => fOffset != idOffset, out count);
                    int tempId = Interlocked.Increment(ref TempId);
                    string tempFilePath = filePath + "_" + tempId;
                    if (count > 0 && otherContent != null)
                    {
                        using (FileStream fs = new FileStream(tempFilePath, FileMode.Append))
                        {
                            fs.Write(otherContent, 0, count);
                        }
                    }
                    SaveValues(id, tempFilePath, values);
                    DiskVisitor.SafeDeleteFile(filePath);
                    DiskVisitor.SafeMoveFile(tempFilePath, filePath);
                }
            });
        }

        public List<SmallValue16AndFlag> GetValues(int id)
        {
            string filePath = GetFilePath(id);
            byte idOffset = (byte)(id % combineRate);
            int count = 0;
            byte[] idContent = GetFileContent(filePath, (fOffset) => fOffset == idOffset, out count);
            if (idContent == null || count <= 0)
            {
                return null;
            }
            List<SmallValue16AndFlag> result = new List<SmallValue16AndFlag>(count / CellLen);
            byte[] idBuffer = new byte[SmallValue16AndFlag.CellLen];
            for (int cI = 0; cI < count; cI += CellLen)
            {
                Array.Copy(idContent, cI + 1, idBuffer, 0, SmallValue16AndFlag.CellLen);
                result.Add(SmallValue16AndFlag.Parse(idBuffer, 0));
            }
            return result;
        }

        internal byte[] GetFileContent(string filePath, Func<byte, bool> filter, out int count)
        {
            if (!File.Exists(filePath))
            {
                count = 0;
                return null;
            }
            byte[] outArray = null;
            byte[] btArray = null;
            ClusterContext.StringLock.Invoke(filePath, () => {
                using (FileStream fs = new FileStream(filePath, FileMode.Open))
                {
                    if (fs.Length > MaxBufferSize)
                    {
                        throw new NotSupportedException();
                    }
                    btArray = new byte[fs.Length];
                    fs.Read(btArray, 0, btArray.Length);
                }
            });
            outArray = new byte[btArray.Length];
            int outI = 0;
            for (int loopI = 0; loopI < btArray.Length; loopI += CellLen)
            {
                if (filter(btArray[loopI]))
                {
                    Array.Copy(btArray, loopI, outArray, outI, CellLen);
                    outI += CellLen;
                }
            }
            count = outI;
            return outArray;
        }

        public bool Exists(int id)
        {
            string filePath = GetFilePath(id);
            byte idOffset = (byte)(id % combineRate);
            int count = 0;
            byte[] idContent = GetFileContent(filePath, (fOffset) => fOffset == idOffset, out count);
            return idContent != null && count > 0;
        }

        public void TryRemove(List<int> idList)
        {
            foreach (var group in idList.GroupBy(id => id / combineRate))
            {
                string filePath = getIdPathFunc(group.Key);
                ClusterContext.StringLock.Invoke(filePath, () => {
                    if (File.Exists(filePath))
                    {
                        byte[] idOffsetArray = group.Select(id => (byte)(id % combineRate)).ToArray();
                        int count = 0;
                        byte[] idContent = GetFileContent(filePath, (fOffset) => !idOffsetArray.Contains(fOffset), out count);
                        if (idContent != null && count > 0)
                        {
                            int tempId = Interlocked.Increment(ref TempId);
                            string tempFilePath = filePath + "_" + tempId;
                            using (FileStream fs = new FileStream(tempFilePath, FileMode.Append))
                            {
                                fs.Write(idContent, 0, count);
                            }
                            DiskVisitor.SafeDeleteFile(filePath);
                            DiskVisitor.SafeMoveFile(tempFilePath, filePath);
                        }
                        else
                        {
                            DiskVisitor.SafeDeleteFile(filePath);
                        }
                    }
                });
            }
        }

        internal void VisitBatchDataAdapt(List<string> filePathList, Action<List<KeyValuePair<int, List<SmallValue16AndFlag>>>> action)
        {
            List<KeyValuePair<int, SmallValue16AndFlag>> kvList = new List<KeyValuePair<int, SmallValue16AndFlag>>();
            foreach (var currentPath in filePathList)
            {
                string fileName = currentPath.Substring(currentPath.LastIndexOf("\\") + 1);
                if (fileName.IndexOf("_") >= 0)
                {
                    continue;
                }
                int baseI = int.Parse(fileName) * combineRate;
                int count = 0;
                byte[] btArray = GetFileContent(currentPath, (fOffset) => true, out count);
                if (btArray != null && count > 0)
                {
                    for (int loopI = 0; loopI < count; loopI += CellLen)
                    {
                        kvList.Add(new KeyValuePair<int, SmallValue16AndFlag>((int)(btArray[loopI]) + baseI, SmallValue16AndFlag.Parse(btArray, loopI + 1)));
                    }
                }
            }
            List<KeyValuePair<int, List<SmallValue16AndFlag>>> paramList = kvList.GroupBy(item => item.Key).Select(group => new KeyValuePair<int, List<SmallValue16AndFlag>>(group.Key, group.Select(groupItem => groupItem.Value).ToList())).ToList();
            action(paramList);
        }

        public void VisitAll(int batchSize, Action<List<KeyValuePair<int, List<SmallValue16AndFlag>>>> action)
        {
            List<string> filePathList = new List<string>(batchSize);
            DiskVisitor.SequenceVisitFiles(basePath, (filePath) => {
                filePathList.Add(filePath);
                if (filePathList.Count >= batchSize)
                {
                    VisitBatchDataAdapt(filePathList, action);
                    filePathList.Clear();
                }
                return false;
            });
            if (filePathList.Count > 0)
            {
                VisitBatchDataAdapt(filePathList, action);
            }
        }
    }

}
